package com.wei.outputformat;

import java.io.IOException;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

public class MyRecordWriter extends RecordWriter<Text, NullWritable>{
	FSDataOutputStream enhanceOutputStream=null;
	FSDataOutputStream toCrawlputStream=null;
	
	

	public MyRecordWriter(FSDataOutputStream enhanceOutputStream,FSDataOutputStream toCrawlputStream) {
		this.enhanceOutputStream = enhanceOutputStream;
		this.toCrawlputStream = toCrawlputStream;
	}

	@Override
	public void write(Text key, NullWritable value) throws IOException,
			InterruptedException {
		if (key.toString().endsWith("toCrawl")) {
			toCrawlputStream.write(key.getBytes());
		}else{
			enhanceOutputStream.write(key.getBytes());
		}
		
	}

	@Override
	public void close(TaskAttemptContext context) throws IOException,
			InterruptedException {
		if (enhanceOutputStream != null) {
			enhanceOutputStream.close();
		}
		if (toCrawlputStream != null) {
			toCrawlputStream.close();
		}
	}

}
